package com.oildatacollection.datasend;

import com.oildatacollection.message.SendMessage;
import com.oildatacollection.message.protocol.MessageProtocol;
import com.oildatacollection.message.protocol.MessageProtocolModbusImpl;
import com.oildatacollection.util.YamlLoader;
import io.netty.buffer.ByteBuf;
import io.netty.channel.socket.SocketChannel;
import lombok.extern.slf4j.Slf4j;
import java.util.Hashtable;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
public class SendQueueImpl implements SendQueue {
    private LinkedBlockingQueue<SendMessage> lbq;
    private final Map<Integer, SocketChannel> chMap;
    private final AtomicBoolean stoped;
    private Thread sendThread;
    private final MessageProtocol msgPro;

    public SendQueueImpl() {
        lbq = null;
        chMap = new Hashtable<>();
        stoped = new AtomicBoolean(true);
        sendThread = null;
        msgPro = new MessageProtocolModbusImpl();
    }

    private void initConfig() {
        try {
            YamlLoader yamlLoader = new YamlLoader("app-config/application-server.yml");
            Map<String, Object> ymlObj = (Map<String, Object>) yamlLoader.get("dataSend");
            int capacity = (int) ymlObj.get("queueCapacity");
            lbq = new LinkedBlockingQueue<>(capacity);
        } catch (RuntimeException e) {
            lbq = new LinkedBlockingQueue<>(100);
            log.error("SendQueue: 找不到配置文件", e);
        }
        log.trace("发送队列初始化完毕");
    }

    private class QueThread implements Runnable {
        @Override
        public void run() {
            while (!stoped.get()) {
                try {
                    // 获取消息
                    SendMessage msg = lbq.take();
                    // 判断是否为关闭消息
                    if (msg.isEmpty()) {
                        break;
                    }
                    // 查找相应的SocketChannel
                    int wellID = msg.getWellID();
                    SocketChannel ch = chMap.get(wellID);
                    if (ch == null) {
                        log.trace("无法发送控制数据: wellID对应的socket不存在");
                        continue;
                    }
                    // 如果为关闭消息 则对套接字进行关闭
                    if (msg.isClose()) {
                        ch.close();
                    } else {
                        // 对数据进行编码
                        ByteBuf res = msgPro.incodeMessage(msg, (byte) 0x10);
                        // 发送数据
                        ch.writeAndFlush(res);
                        log.trace("控制数据发送完毕");
                    }
                } catch (InterruptedException e) {
                    log.error("消息发送失败", e);
                }
            }
            log.trace("发送队列终止运行");
        }
    }

    @Override
    public void sendToQueue(SendMessage sdMsg) {
        try {
            lbq.put(sdMsg);
        } catch (InterruptedException e) {
            log.error("sendToQueue错误", e);
        }
    }

    @Override
    public void start() {
        initConfig();
        stoped.set(false);
        sendThread = new Thread(new QueThread(), "SENDER_QUE_THREAD");
        sendThread.start();
        log.trace("发送队列模块启动成功");
    }

    @Override
    public void stop() {
        stoped.set(true);
        sendToQueue(new SendMessage(0, new byte[0]));
        try {
            sendThread.join();
        } catch (InterruptedException e) {
            log.error("SendQue::stop", e);
        }
    }

    @Override
    public void addAddress(int wellID, SocketChannel ch) {
        chMap.put(wellID, ch);
    }
}
